
'''
MQTT_RECEIVE_QUEUE 队列用于存储接收到的MQTT消息

从以上队列中获取MQTT消息并处理
'''

import redis
import threading
import json
from utils.custom_logger import CustomLogger
from db_utils.postgresql_connection import PostgreSQLConnection
from db_utils.repositories.device_data_table import DeviceDataTable
from settings import MQTT_RECEIVE_QUEUE, MQTT_HOST, MQTT_PORT, MQTT_CLIENT_CERT, MQTT_AUTH, MQTT_PASSWORD, MQTT_TOPIC, REDIS_HOST, REDIS_PORT, REDIS_DB


class IotMqttSubscriberWorker:

    logger = CustomLogger(logger_name='iot_mqtt_subscriber_worker', log_file_path='logs/iot_mqtt_subscriber_worker.log').get_logger()

    def __init__(self, worker_num=2):

        # worker数量
        self.worker_num = worker_num

        # worker 线程列表
        self.worker_threads = []


    def worker(self):

        redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)

        while True:

            try:
                # 从 MQTT_RECEIVE_QUEUE 队列中获取消息
                message_raw = redis_client.blpop(MQTT_RECEIVE_QUEUE, timeout=20) # 添加超时时间，防止redis挂掉导致无限等待

                if not message_raw:

                    continue # 空消息，跳过

            except redis.RedisError as e:

                self.logger.error(f"redis connection lost: {e}")

                redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB) # 重新连接redis
                continue

            except Exception as e:

                self.logger.error(f"redis connection error: {e}")
                continue

            else:
                
                self.logger.debug(f"线程 {threading.current_thread().name} 从 MQTT_RECEIVE_QUEUE 队列中获取消息成功: {message_raw[1]}")


            try:
                message = json.loads(message_raw[1])  # message:  {"type": "test_message", "data": {"device_code": "1234567890", "message": "send to redis"}}

                with PostgreSQLConnection() as (conn, cursor):
                    
                    device_data_table = DeviceDataTable(conn, cursor)
                    # def create_device_data(self, device_id, data_type, data_json, data_direction):

                    device_code = message['data']['device_code']
                    data_type = message['type']
                    # data_json = message['data'] # 不需要json.dumps() 可以是dict类型
                    data_json = json.dumps(message['data']) # 转换为json字符串
                    data_direction = 'up' # 上传数据

                    insert_result = device_data_table.create_device_data_by_device_code(device_code, data_type, data_json, data_direction)

            except Exception as e:
                self.logger.error(f"MQTTReceiveQueueWorker worker error: {e}")

            else:

                if insert_result:
                    self.logger.info(f"线程 {threading.current_thread().name} 处理 MQTT 消息成功: {message}")
                else:
                    self.logger.error(f"线程 {threading.current_thread().name} 处理 MQTT 消息失败: {message}")
                

    def start(self):

        # 启动 worker 线程
        for i in range(self.worker_num):
            t = threading.Thread(target=self.worker, name=f"worker_{i}", daemon=True)
            t.start()
            self.worker_threads.append(t)

        # 等待 worker 线程结束
        for t in self.worker_threads:
            t.join()

    
if __name__ == '__main__':
    IotMqttSubscriberWorker().start()
    

                

